package com.rem.kafka.consumer;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 定义某个分区的kafka消费者
 *
 * @author Rem
 * @date 2023-01-03
 */
public class CustomConsumerPartition {
    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger root = loggerContext.getLogger("root");
        root.setLevel(Level.INFO);
    }

    public static void main(String[] args) {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");
        //消费组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group02");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //创建消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties, new StringDeserializer(), new StringDeserializer());

        /**
         * 手动为这个消费者分配一个分区列表。此接口不允许增量赋值，并将替换之前的赋值（如果有的话）。
         * 如果给定的主题分区列表为空，则将其视为与unsubscribe()相同。
         * 通过此方法手动分配主题不会使用消费者的组管理功能。因此，当组成员或集群和主题元数据发生变化时，不会触发重新平衡操作。
         * 请注意，不可能同时使用assign(Collection)的手动分区分配和subscribe(Collection, ConsumerRebalanceListener)的组分配。
         */
        kafkaConsumer.assign(Collections.singletonList(new TopicPartition("first", 0)));

        //消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(System.err::println);
        }
    }
}
